package org.llc.flink.wordcount.api.sinkTest

import org.apache.flink.api.common.serialization.SimpleStringEncoder
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink
import org.apache.flink.streaming.api.scala._
import org.llc.flink.wordcount.api.SensorReading


object FileSink {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

//    env.setParallelism(1)

    // 读取数据
    val path = this.getClass.getClassLoader.getResource("sensor.txt").getPath
    val inputStream: DataStream[String] = env.readTextFile(path)

    // 基本转换
    val dataStream: DataStream[SensorReading] = inputStream
      .map( line => {
        val arr = line.split(",")
        SensorReading(arr(0).trim, arr(1).trim.toLong, arr(2).trim.toDouble)
      } )

    val outPath = this.getClass.getClassLoader.getResource("out.txt").getPath
    // 写入文件
    dataStream.addSink(
      StreamingFileSink.forRowFormat(
        new Path(outPath),
        new SimpleStringEncoder[SensorReading]("UTF-8")
      ).build()
    )

    env.execute("file sink job")
  }
}
